package com.sfx.shortLink.rmq.consumer;


import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.sfx.shortLink.common.convention.exception.ServiceException;
import com.sfx.shortLink.dto.ShortLinkStatsRecordDTO;
import com.sfx.shortLink.mq.idempotent.MessageIdempotentHandler;
import com.sfx.shortLink.rmq.config.RocketMQConsumerProperties;
import com.sfx.shortLink.service.impl.ShortLinkServiceImpl;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

import static com.sfx.shortLink.rmq.topic.TopicConstants.DELAY_QUEUE_STATS_KEY;

@RequiredArgsConstructor
@Slf4j
@Component
public class ShortLinkRmqStatsDelayQueueConsumer implements InitializingBean {

    private static final AtomicInteger index = new AtomicInteger();

    private final MessageIdempotentHandler messageIdempotentHandler;
    private final ShortLinkServiceImpl shortLinkService;
    private final RocketMQConsumerProperties rocketMQConsumerProperties;

    public void onMessage() throws MQClientException {
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer();
        //老版本中会开启，新版本的mq不需要使用到
        mqPushConsumer.setVipChannelEnabled(false);
        mqPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameSrv());
        mqPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupName() + "_" + ShortLinkRmqStatsDelayQueueConsumer.class.getSimpleName());
        //一次从broker中拉取10条消息到本地内存当中进行消费
        mqPushConsumer.setConsumeMessageBatchMaxSize(10);
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //监听im发送过来的业务消息topic
        mqPushConsumer.subscribe(DELAY_QUEUE_STATS_KEY, "*");
        mqPushConsumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                Executors.newSingleThreadExecutor(r -> {
                    Thread t = new Thread(r);
                    t.setName("ShortLinkRmqStatsDelayQueueConsumer-" + index.getAndDecrement());
                    t.setDaemon(Boolean.TRUE);
                    return t;
                }).execute(() -> {
                    for (; ; ) {
                        try {
                            Map<String, String> producerMap = JSON.parseObject(new String(msg.getBody()), Map.class);
                            ShortLinkStatsRecordDTO statsRecordDTO = JSONUtil.parse(producerMap.get("statsRecordDTO")).toBean(ShortLinkStatsRecordDTO.class);
                            String keys = Objects.requireNonNull(statsRecordDTO).getKeys();
                            if (!messageIdempotentHandler.isConsume(keys)) {
                                if (messageIdempotentHandler.isAccomplish(keys)) {
                                    break;
                                }
                                throw new ServiceException("消息消费处理出现异常,无法被消费");
                            }
                            try {
                                shortLinkService.shortLinkStats(statsRecordDTO.getFullShortUrl(), null, statsRecordDTO);
                            } catch (Throwable throwable) {
                                messageIdempotentHandler.delMessage(keys);
                                log.error("延迟消费处理异常", throwable);
                            }
                            // 消息消费已完成
                            messageIdempotentHandler.setAccomplish(keys);
                            continue;
                        } catch (Throwable throwable) {
                        }
                        LockSupport.parkUntil(500);
                    }
                });
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        mqPushConsumer.start();
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        onMessage();
    }
}
